package org.example.elastic.lite.handler;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
import org.apache.shardingsphere.elasticjob.dataflow.job.DataflowJob;
import org.apache.shardingsphere.elasticjob.dataflow.props.DataflowJobProperties;
import org.apache.shardingsphere.elasticjob.infra.pojo.JobConfigurationPOJO;
import org.apache.shardingsphere.elasticjob.infra.yaml.YamlEngine;
import org.apache.shardingsphere.elasticjob.lite.api.bootstrap.impl.OneOffJobBootstrap;
import org.apache.shardingsphere.elasticjob.lite.api.bootstrap.impl.ScheduleJobBootstrap;
import org.apache.shardingsphere.elasticjob.lite.internal.schedule.JobRegistry;
import org.apache.shardingsphere.elasticjob.lite.internal.storage.JobNodePath;
import org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperRegistryCenter;
import org.apache.shardingsphere.elasticjob.simple.job.SimpleJob;
import org.example.elastic.lite.config.DynamicElasticJobListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * 动态定时任务相关操作工具类
 * @author 魏荣杰
 * @date 2022/11/11 10:02
 * @since
 * @version
 */
@Slf4j
@Component
public class DynamicElasticJobHandler {

    @Resource
    private ZookeeperRegistryCenter zookeeperRegistryCenter;
    @Resource
    private DynamicElasticJobListener dynamicElasticJobListener;
//    @Resource
//    private JobScheduler jobScheduler;
//    @Resource
//    private JobEventConfiguration jobEventConfiguration;


//    TracingConfiguration<DataSource> tracingConfig = new TracingConfiguration<>("RDB", setUpEventTraceDataSource());

//    private static DataSource setUpEventTraceDataSource() {
//        BasicDataSource result = new BasicDataSource();
//        result.setDriverClassName(EVENT_RDB_STORAGE_DRIVER);
//        result.setUrl(EVENT_RDB_STORAGE_URL);
//        result.setUsername(EVENT_RDB_STORAGE_USERNAME);
//        result.setPassword(EVENT_RDB_STORAGE_PASSWORD);
//        return result;
//    }


    /**
     * Description: 添加简单job
     * @param jobName
	 * @param cron
	 * @param shardingTotalCount
	 * @param simpleJob
	 * @param parameters
	 * @param descrption
     * @return void
     * @throws
     * @author 魏荣杰
     * @date 2022/11/17 19:54
     * @since
     */
    public void addSimpleJob(String jobName, String cron, int shardingTotalCount, SimpleJob simpleJob, String parameters, String descrption) {
//        TracingConfiguration tracingConfig = new TracingConfiguration()

//        TracingConfiguration<DynamicElasticJobListener> tracingConfiguration = new TracingConfiguration<>(, dynamicElasticJobListener);

//        JobInstance jobInstance = new JobInstance();
//        StdScheduler stdScheduler = new StdScheduler();
//        JobDetail jobDetail = JobBuilder.newJob(LiteJob.class).withIdentity(jobName).build();
//        JobScheduleController jobScheduleController = new JobScheduleController(stdScheduler, jobDetail, jobName);

//        JobScheduler jobScheduler = new JobScheduler();
//        JobRegistry.getInstance().registerJob(jobName, jobScheduleController);

//        addSimpleJobSingle(jobName, cron, shardingTotalCount, simpleJob, parameters, descrption);
        addSimpleJobBatch(jobName, cron, shardingTotalCount, simpleJob, parameters, descrption);
    }

    private void addSimpleJobSingle(String jobName, String cron, int shardingTotalCount, SimpleJob simpleJob, String parameters, String descrption) {
        JobConfiguration jobConfiguration = JobConfiguration.newBuilder(jobName, shardingTotalCount)
                .cron(cron)
                .shardingItemParameters(parameters)
                .description(descrption)
                .timeZone("GMT+8:00")
                .overwrite(true)
                .jobListenerTypes(dynamicElasticJobListener.getType())
//                .addExtraConfigurations()
//                .jobExecutorServiceHandlerType()
//                .monitorExecution()
//                .jobParameter()
//                .addExtraConfigurations(tracingConfig)
                .build();
        new ScheduleJobBootstrap(zookeeperRegistryCenter, simpleJob, jobConfiguration).schedule();
    }

    private void addSimpleJobBatch(String jobName, String cron, int shardingTotalCount, SimpleJob simpleJob, String parameters, String descrption) {
        Runnable runnable = () -> {
            for (int i = 1; i <= 10000; i++) {
                String name = jobName + i;
                JobConfiguration jobConfiguration = JobConfiguration.newBuilder(name, shardingTotalCount)
                        .cron(cron)
                        .shardingItemParameters(parameters)
                        .description(descrption)
                        .timeZone("GMT+8:00")
                        .overwrite(true)
                        .jobListenerTypes(dynamicElasticJobListener.getType())
//                .addExtraConfigurations()
//                .jobExecutorServiceHandlerType()
//                .monitorExecution()
//                .jobParameter()
//                .addExtraConfigurations(tracingConfig)
                        .build();
                new ScheduleJobBootstrap(zookeeperRegistryCenter, simpleJob, jobConfiguration).schedule();
            }
            log.info("addSimpleJobBatch finish!!!!!");
        };
        Thread thread = new Thread(runnable);
        thread.start();
    }

    /**
     * Description: 添加流job
     * @param jobName
	 * @param cron
	 * @param shardingTotalCount
	 * @param dataflowJob
	 * @param parameters
	 * @param descrption
     * @return void
     * @throws
     * @author 魏荣杰
     * @date 2022/11/17 19:55
     * @since
     */
    public void addDataflowJob(String jobName, String cron, int shardingTotalCount, DataflowJob<?> dataflowJob, String parameters, String descrption) {
        addDataflowJobSingle(jobName, cron, shardingTotalCount, dataflowJob, parameters, descrption);
//        addDataflowJobBatch(jobName, cron, shardingTotalCount, dataflowJob, parameters, descrption);
    }

    private void addDataflowJobSingle(String jobName, String cron, int shardingTotalCount, DataflowJob<?> dataflowJob, String parameters, String descrption) {
        JobConfiguration jobConfiguration = JobConfiguration.newBuilder(jobName, shardingTotalCount)
                .cron(cron)
                .shardingItemParameters(parameters)
                .setProperty(DataflowJobProperties.STREAM_PROCESS_KEY, Boolean.TRUE.toString())
                .overwrite(true)
                .description(descrption)
//                .addExtraConfigurations(tracingConfig)
                .build();

        new ScheduleJobBootstrap(zookeeperRegistryCenter, dataflowJob, jobConfiguration).schedule();
    }
    private void addDataflowJobBatch(String jobName, String cron, int shardingTotalCount, DataflowJob<?> dataflowJob, String parameters, String descrption) {
        Runnable runnable = () -> {
            for (int i = 1; i <= 10000; i++) {
                String name = jobName + i;
                JobConfiguration jobConfiguration = JobConfiguration.newBuilder(name, shardingTotalCount)
                        .cron(cron)
                        .shardingItemParameters(parameters)
                        .setProperty(DataflowJobProperties.STREAM_PROCESS_KEY, Boolean.TRUE.toString())
                        .overwrite(true)
                        .description(descrption)
//                .addExtraConfigurations(tracingConfig)
                        .build();

                new ScheduleJobBootstrap(zookeeperRegistryCenter, dataflowJob, jobConfiguration).schedule();
            }
            log.info("addDataflowJobBatch finish!!!!!");
        };
        Thread thread = new Thread(runnable);
        thread.start();
    }

    /**
     * Description: 添加单次执行的简单job
     * @param jobName
	 * @param cron
	 * @param shardingTotalCount
	 * @param simpleJob
	 * @param parameters
	 * @param descrption
     * @return void
     * @throws
     * @author 魏荣杰
     * @date 2022/11/17 19:55
     * @since
     */
    public void addOneOffSimpleJob(String jobName, String cron, int shardingTotalCount, SimpleJob simpleJob, String parameters, String descrption) {
        JobConfiguration jobConfiguration = JobConfiguration.newBuilder(jobName, shardingTotalCount)
                .cron(cron)
                .shardingItemParameters(parameters)
                .description(descrption)
                .overwrite(true)
//                .addExtraConfigurations(tracingConfig)
                .build();
        new OneOffJobBootstrap(zookeeperRegistryCenter, simpleJob, jobConfiguration).execute();
    }



    public void updateJob(String jobName, String cron) {
        log.info("update job, jobName: {}, cron: {}", jobName, cron);
//        jobScheduler.getJobScheduleController().rescheduleJob(cron, "GMT+8:00");
        // job的更新
        JobRegistry.getInstance().getJobScheduleController(jobName).rescheduleJob(cron, "GMT+8:00");

        // zookeeper的更新
        JobNodePath jobNodePath = new JobNodePath(jobName);
        String yamlContent = zookeeperRegistryCenter.get(jobNodePath.getConfigNodePath());
        JobConfigurationPOJO jobConfigurationPOJO = null == yamlContent ? null : (JobConfigurationPOJO) YamlEngine.unmarshal(yamlContent, JobConfigurationPOJO.class);
        if (Objects.nonNull(jobConfigurationPOJO)) {
            jobConfigurationPOJO.setCron(cron);
            jobConfigurationPOJO.setJobName(jobName);
            zookeeperRegistryCenter.update(jobNodePath.getConfigNodePath(), YamlEngine.marshal(jobConfigurationPOJO));
        }
    }

    public void removeJob(String jobName) {
        log.info("remove job, jobName: {}", jobName);
//        removeJobSingle(jobName);
        removeJobBatch(jobName);
    }

    private void removeJobSingle(String jobName) {
//        jobScheduler.getJobScheduleController().shutdown();

        JobRegistry.getInstance().getJobScheduleController(jobName).shutdown();
        zookeeperRegistryCenter.remove("/" + jobName);
    }
    private void removeJobBatch(String jobName) {
        Runnable runnable = () -> {
            for (int i = 1; i <= 3000; i++) {
                String name = jobName + i;
                try {
                    JobRegistry.getInstance().getJobScheduleController(jobName).shutdown();
                } catch (Exception e) {
                    log.error("job shutdown fail", e);
                }
                zookeeperRegistryCenter.remove("/" + name);
            }
            log.info("removeJobBatch finsih!");
        };
        Thread thread = new Thread(runnable);
        thread.start();
    }

    public static void main(String[] args) {
        String s = "";
        List<String> strings = JSON.parseArray(s, String.class);
        Set<String> collect = strings.stream().collect(Collectors.toSet());

        String s1 = "";
        List<String> strings1 = JSON.parseArray(s1, String.class);
        Set<String> collect1 = strings1.stream().collect(Collectors.toSet());

        collect.removeAll(collect1);

        System.out.println(JSON.toJSONString(collect));
    }
}
